package ws.very.util.akka.cluster

import akka.actor.Stash
import akka.cluster.ClusterEvent._
import akka.cluster.Member
import akka.actor.LoggingFSM
import akka.actor.Address
import akka.actor.ActorRef
import akka.actor.ActorSelection
import ws.very.util.akka.actor.ReplyAct
import akka.util.Timeout
import scala.util.Success
import scala.util.Failure

object ClusterLeaderFSM {
  sealed trait LeaderState
  object Waiting extends LeaderState
  object Ready extends LeaderState
}

trait ClusterLeaderFSM extends ClusterListener with Stash with LoggingFSM[ClusterLeaderFSM.LeaderState, (O[Address], Set[Member])] { //akka.actor.debug.fsm open log
  import ClusterLeaderFSM._
  override def logDepth = 12
  startWith(Waiting, None -> Set.empty[Member])

  //  def roleFilter:O[S=>B]=None FIXME:
  def isSelfLeader(curLeader: Address) = cluster.selfAddress == curLeader
  def listen = Seq(classOf[MemberUp], classOf[LeaderChanged], classOf[MemberRemoved])

  def whenWaiting: StateFunction = {
    case Event(LeaderChanged(Some(addr)), curData) =>
      unstashAll()
      goto(Ready) using (Some(addr) -> curData._2)
    case other =>
      stash()
      stay
  }
  def whenReady: StateFunction = {
    case Event(LeaderChanged(None), curData) =>
      goto(Waiting) using (None -> curData._2)
    case Event(LeaderChanged(Some(addr)), curData) =>
      stay using (Some(addr) -> curData._2)
    case Event(c: CurrentClusterState, curData) =>
      stay using (curData._1 -> c.members)
    case Event(MemberUp(m), curData) =>
      stay using (curData._1 -> (curData._2 + m))
    case Event(MemberRemoved(m, _), curData) =>
      stay using (curData._1 -> (curData._2 - m))

  }
  when(Waiting) { whenWaiting }

  when(Ready) { whenReady }

  override def preStart = {
    super.preStart()
    initialize
  }
}
object ClusterReplication {
  trait ReplicationAction {
    def leaderAction: B
  }
  case class Send2All(msg: A, leaderAction: B = true) extends ReplicationAction
  import scala.concurrent.duration._
  implicit val askTimeOut = Timeout(2 seconds)
}

trait ClusterReplication extends ClusterLeaderFSM with ReplyAct {
  import ClusterLeaderFSM._
  import ClusterReplication._
  import akka.pattern.ask

  override def whenReady = super.whenReady.orElse {
    case Event(send @ Send2All(msg, leaderAction), (Some(leader), members)) =>
      if (leaderAction && isSelfLeader(leader)) {
        log.info(s"leader send2all $msg")
        members.foreach { m => addrRef(m.address) ask msg }
        reply(true)
      } else {
        log.info(s"forward 2 leader  $msg")
        addrRef(leader) forward (send)
      }
      stay
  }

  def addrRef(addr: Address): ActorSelection

  //  abstract override def receive = {
  //    case c: CurrentClusterState =>
  //      members = c.members
  //      unstashAll()
  //      context.become(clusterListen.orElse { super.receive })
  //    case other => stash
  //  }

  //  def masterWhenNewMember(member: Member)
  //
  //  def forward2Master(msg: A)
  //
  //  def clusterListen: Pfn[A, U] = {
  //
  //    case MemberUp(member) =>
  //      members = members + member
  //      if (isSelfLeader && member.address != cluster.selfAddress)
  //        masterWhenNewMember(member)
  //
  //    case all @ Send2All(msg) =>
  //      if (!isSelfLeader)
  //        forward2Master(all)
  //
  //    //      log.info(s"listen reg=$reg")
  //    //      if (rightReg(reg)) {
  //    //        super.add2Route(reg)
  //    //        if (isSelfLeader)
  //    //          members /*FIXME:不同步*/ .filter { member => member.address != cluster.selfAddress }.map { _.address }.foreach { add =>
  //    //            remoteRefSend(add, ClusterMsgCenter.SyncRegForward(reg))
  //    //          }
  //    //      }
  //  }

}